package com.xing.disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;

import java.util.concurrent.*;

public class DisruptorExecutor implements Executor {
	
	private Disruptor<DisruptorEvent> disruptor;
	private WorkerPool<DisruptorEvent> workerPool;
	private RingBuffer<DisruptorEvent> ringBuffer;
	private ExecutorService exec;
	
	public DisruptorExecutor(final String name, int bufferSize) {
		if(!is2XValue(bufferSize)) {
			throw new IllegalArgumentException("bufferSize必须是2的n次方");
		}
		exec = name == null ? Executors.newSingleThreadExecutor() : Executors.newSingleThreadExecutor(new ThreadFactory() {
			public Thread newThread(Runnable r) {
				Thread t = new Thread(r, name);
				t.setPriority(Thread.MAX_PRIORITY);
				return t;
			}
		});
		workerPool = null;
		disruptor = new Disruptor<DisruptorEvent>(DisruptorEvent.EVENT_FACTORY, bufferSize, exec);
		EventHandler<DisruptorEvent>[] handlers = new DisruptorEventHandler[1];
		handlers[0] = new DisruptorEventHandler();
		disruptor.handleEventsWith(handlers);
		ringBuffer = disruptor.start();
	}
	
	public DisruptorExecutor(final String name, int num, int bufferSize) {
		if(!is2XValue(bufferSize)) {
			throw new IllegalArgumentException("bufferSize必须是2的n次方");
		}
		exec = name == null ? Executors.newFixedThreadPool(num) : Executors.newFixedThreadPool(num, new ThreadFactory() {
			public Thread newThread(Runnable r) {
				Thread t = new Thread(r, name);
				t.setPriority(Thread.MAX_PRIORITY);
				return t;
			}
		});
		disruptor = null;
		WorkHandler<DisruptorEvent>[] handlers = new DisruptorEventHandler[num];
		for(int idx = 0; idx < num; idx ++) {
			handlers[idx] = new DisruptorEventHandler();
		}
		workerPool = new WorkerPool<DisruptorEvent>(DisruptorEvent.EVENT_FACTORY, new ExceptionHandler() {
			public void handleOnStartException(Throwable ex) {
				ex.printStackTrace();
			}
			
			public void handleOnShutdownException(Throwable ex) {
				ex.printStackTrace();
			}
			
			public void handleEventException(Throwable ex, long sequence, Object event) {
				ex.printStackTrace();
			}
		}, handlers);
		ringBuffer = workerPool.start(exec);
	}
	
	private boolean is2XValue(int value) {
		return value > 0 && (value & (value - 1)) == 0;
	}
	
	public void execute(Runnable task) {
		long sequence = ringBuffer.next();
		try {
			DisruptorEvent event = ringBuffer.get(sequence);
			event.task = task;
		} finally {
			ringBuffer.publish(sequence);
		}
	}
	
	public void shutdown() {
		if(disruptor == null) {
			workerPool.drainAndHalt();
		} else {
			disruptor.shutdown();
		}
		exec.shutdown();
		try {
			exec.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	private static class DisruptorEvent {
		public static final EventFactory<DisruptorEvent> EVENT_FACTORY = new EventFactory<DisruptorEvent>() {
			public DisruptorEvent newInstance() {
				return new DisruptorEvent();
			}
		};
		
		public Runnable task;
		
		public void run() {
			try {
				task.run();
			} catch (Throwable e) {
				e.printStackTrace();
			}
		}
	}
	
	private static class DisruptorEventHandler implements EventHandler<DisruptorEvent>, WorkHandler<DisruptorEvent> {
		public void onEvent(DisruptorEvent event) throws Exception {
			event.run();
		}
		
		public void onEvent(DisruptorEvent event, long sequence, boolean endOfBatch) throws Exception {
			event.run();
		}
	}
}
